
#include <string>
#include <sstream>
#include "NetworkFw.h"

FwNetworking::FwNetworking(AppNode *an_app): app(an_app), listen_sock(0), epollfd(0) {

    if ((epollfd = epoll_create(TNMC_MAX_EPOLL_SIZE)) == -1) {
        logger.log("VSS epoll creation failed!", LOG_LEVEL_FATAL);
        exit(1);
    }
}

FwNetworking::~FwNetworking() {
    if (listen_sock) delete listen_sock;
}

int FwNetworking::PeekBlock(int sockfd, char *buf, size_t len) throw (int)
{
    errno = 0;
    int recvlen = recv(sockfd, buf, len, MSG_PEEK);
    if (recvlen <= 0) {
        if (recvlen == 0) {
            throw TNMC_FW_EVENT_CODE_CLOSE; 
        } else if (errno == EAGAIN) {
            throw TNMC_FW_EVENT_CODE_EGAIN;
        } else {
            throw TNMC_FW_EVENT_CODE_NETWORK_ERR;
        }
    }

    buf[recvlen] = '\0';
    return recvlen;
}

int FwNetworking::ReadBlock(int sockfd, char *buf, size_t len) throw(int)
{
    errno = 0;
    int recvlen = recv(sockfd, buf, len, 0);
    if (recvlen <= 0) {
        if (recvlen == 0) {
            throw TNMC_FW_EVENT_CODE_CLOSE; 
        } else if (errno == EAGAIN) {
            throw TNMC_FW_EVENT_CODE_EGAIN;
        } else {
            throw TNMC_FW_EVENT_CODE_NETWORK_ERR;
        }
    }

    buf[recvlen] = '\0';
    app->AddByteCount((uint64_t)recvlen, true);
    return recvlen;
}

int FwNetworking::SendFileBlock(int sockfd, char *buf, size_t len)
{
    errno = 0;
    int send_res = send(sockfd, buf, len, 0);
    if(send_res <= 0) {
        if (errno == EAGAIN) {
            return 0;
        } else {
            logger.log("VSS socket sending error", LOG_LEVEL_ERROR);
            return -1;
        }
    }
    
    app->AddByteCount((uint64_t)send_res, false);
    return send_res;
}

int FwNetworking::SendString(int sockfd, string datagram) {
    const char *fw_send_buf = datagram.c_str();
    size_t len = datagram.size();

    size_t send_count = 0;
    int try_count = 0;
    while(send_count < len) {
        errno = 0;
        int send_res = send(sockfd, fw_send_buf + send_count, len - send_count, 0);
        if(send_res <= 0) { //failed to send any bytes
            if (errno == EAGAIN) {
                if (try_count++ > 10) {
                    return -1;
                } else { //try again
                    continue;
                }
            } else { //other type of failure
                return -1;
            }
        } else { // bytes sent
            send_count += send_res;
        }
    }

    app->AddByteCount((uint64_t)len, false);
    return 0;
}

int FwNetworking::StartListening(int port)
{
    int ret;//return value
    struct sockaddr_in serveraddr;

    //create TCP socket
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if(listen_fd == -1) {
        logger.log("VSS socket creation failed!", LOG_LEVEL_FATAL);
        return -1;
    }

    SetNonBlocking(listen_fd);
    int one = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));

    memset(&serveraddr, 0, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_port = htons(port);
    serveraddr.sin_addr.s_addr = INADDR_ANY;

    ret = bind(listen_fd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr));
    if (ret == -1) {
        logger.log("VSS socket bind failed!", LOG_LEVEL_FATAL);
        return -1;
    } else {
        logger.log(" server socket create and bind success!", LOG_LEVEL_INFO);
    }

    ret = listen(listen_fd, TNMC_SERVER_LISTEN_SIZE);
    if(ret == -1) {
        logger.log("VSS listen failed!", LOG_LEVEL_FATAL);
        return -1;
    }

    listen_sock = new SockNode(listen_fd);

    struct epoll_event ev;
    ev.events = EPOLLIN | EPOLLET;
    ev.data.ptr = (void *)listen_sock;
    
    ret = epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_fd, &ev);
    if (ret < 0) {
        logger.log("VSS epoll fd setting failed!", LOG_LEVEL_ERROR);
        exit(1);
    }
    
    return 0;
}

int FwNetworking::SetNonBlocking(int sockfd)
{
    int opts = fcntl(sockfd, F_GETFL);
    if (opts < 0) {
        return -1;
    }

    opts = opts|O_NONBLOCK;
    if (fcntl(sockfd, F_SETFL, opts) < 0) {
        return -1;
    }

    return 0;
}

void FwNetworking::GetLocalSockIp(int sockfd, string &local_ip_text)
{
    struct sockaddr_in localaddr;
    socklen_t localaddr_len = sizeof(localaddr);
    if (getsockname(sockfd, (struct sockaddr *)&localaddr, &localaddr_len)) {
        logger.log("Unable to get local address....", LOG_LEVEL_ERROR);
        return;
    }
    
    local_ip_text = inet_ntoa(localaddr.sin_addr);
}

void FwNetworking::AcceptConn() throw (SockNode*) {
    struct sockaddr_in clientaddr;
    socklen_t len = sizeof (struct sockaddr_in);
    int newfd = accept(listen_sock->sockfd, (struct sockaddr*)&clientaddr, &len);
    if (newfd < 0) return;

    SetNonBlocking(newfd);

    SockNode *newsock = new SockNode(newfd);
    struct epoll_event ev;
    ev.events = EPOLLIN | EPOLLET;
    ev.data.ptr = (void *)newsock;

    cout << "Accept sock(fd/node " << newfd << ":" << newsock << ")" << endl;
    int ret = epoll_ctl(epollfd, EPOLL_CTL_ADD, newfd, &ev);
    if (ret < 0) {
        logger.log("VSS epoll add failed!", LOG_LEVEL_ERROR);
        throw newsock;
    }

    logger.log("VSS socket received a new connection", LOG_LEVEL_INFO);
    newsock->data = app->AddSession(newsock); 
}

int FwNetworking::InitConn(uint32_t server_ip, uint16_t server_port) {

    int res; 
    //create TCP socket
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if(fd == -1) {
        return -1;
    }

    SetNonBlocking(fd);
    int one = 1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));

    struct sockaddr_in serveraddr;
    memset(&serveraddr, 0, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_port = htons(server_port);
    serveraddr.sin_addr.s_addr = htonl(server_ip);

    SockNode *sock_node = new SockNode(fd);
    sock_node->data = app->AddSession(sock_node);

    struct epoll_event ev;
    ev.events = EPOLLET | EPOLLERR | EPOLLOUT | EPOLLONESHOT;
    ev.data.ptr = (void *)sock_node;

    cout << "Init sock(fd/node " << fd << ":" << sock_node << ")" << endl;
    res = epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev);
    if (res < 0) {
        close(fd);
        logger.log("VSS epoll add failed!", LOG_LEVEL_ERROR);
        return -1;
    }
    
    errno = 0;
    res = connect(fd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));

    if (res == 0) {
        FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_CONNECTED, sock_node);
        app->AddWork(msg);
    } else {
        if (errno != EINPROGRESS) {
            FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_NETWORK_ERR, sock_node);
            app->AddWork(msg);
            return -1;
        }
        app->AddConnectingSession(sock_node);
    }

    return fd;
}

void FwNetworking::RunEpoll()
{
    struct epoll_event events[TNMC_MAX_EPOLL_SIZE];
    while (true) {
        int nfds = epoll_wait(epollfd, events, TNMC_MAX_EPOLL_SIZE, -1);
        if (nfds == -1) {
            logger.log("VSS epoll fd waiting failed!", LOG_LEVEL_ERROR);
            exit(1);
        }
        
        try {
            for (int i = 0; i < nfds; i++) {
                SockNode *sock_node = (SockNode *)events[i].data.ptr;
                if (!sock_node) continue;
    

                if (sock_node == listen_sock) { //new connection request 
                    AcceptConn();
                }
                else if (events[i].events & (EPOLLHUP | EPOLLRDHUP)) {
                    CloseConn(sock_node);
                    FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_CLOSE, sock_node);
                    app->AddWork(msg);
                } else if (events[i].events & EPOLLERR) {
                    throw sock_node;
                } else if (events[i].events & EPOLLIN) {
                    FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_RECV, sock_node);
                    app->AddWork(msg);
                } else if (events[i].events & EPOLLOUT) {
                    events[i].events = EPOLLIN | EPOLLET | EPOLLERR;
                    int ret = epoll_ctl(epollfd, EPOLL_CTL_MOD, sock_node->sockfd, &events[i]);
                    if (ret < 0) { throw sock_node; }
                    FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_CONNECTED, sock_node);
                    app->AddWork(msg);
                }
            }
        } catch (SockNode *sock_node) {
            cout << "Error: "<< sock_node << endl;
            CloseConn(sock_node);
            FwAppMsg *msg = new FwAppMsg(FW_APP_MSG_TYPE_NETWORK_ERR, sock_node);
            app->AddWork(msg);
        }
    }
}

Session::Session(SockNode *node): sockfd(node->sockfd), sn(node), locked(false) {
    pthread_mutex_init(&mutex, NULL);
}

Session::~Session() {
    pthread_mutex_destroy(&mutex);
}

AppNode::AppNode(uint32_t timeout_usec): networker(this),
    m_bytes_rd(0), m_bytes_wr(0),
    m_bytes_rd_save(0), m_bytes_wr_save(0)
{
    m_timeo_usec = timeout_usec % 1000000;
    m_timeo_sec = timeout_usec / 1000000;
    gettimeofday(&start_time, NULL);
    m_rate_time = start_time;
    set_timeout();
    text_buf = new char[TNMC_HELPER_BUFFER_LEN + 1];
    pthread_mutex_init(&stat_mutex, NULL);
}

AppNode::~AppNode() 
{ 
    pthread_mutex_destroy(&stat_mutex);
    delete text_buf;
}

void AppNode::Respond416(int fd)
{
    static const string http_buf = 
        "HTTP/1.1 416 wrong range\r\n"
        "Server: TnmcAppNode/1.0\r\n"
        "Cache_Control: no-cache\r\n"
        "Content-Type: text/xml\r\n"
        "Content-Length: 0\r\n";

    networker.SendString(fd, http_buf);
}

void AppNode::Respond404(int fd)
{
    static const string http_buf = 
        "HTTP/1.1 404 Not Found\r\n"
        "Server: TnmcAppNode/1.0\r\n"
        "Cache_Control: no-cache\r\n"
        "Content-Type: text/xml\r\n"
        "Content-Length: 0\r\n";

    networker.SendString(fd, http_buf);
}

void AppNode::Respond400(int fd)
{
    static const string http_buf = 
        "HTTP/1.1 400 Bad Request\r\n"
        "Server: TnmcAppNode/1.0\r\n"
        "Cache_Control: no-cache\r\n"
        "Content-Type: text/xml\r\n"
        "Content-Length: 0\r\n";

    networker.SendString(fd, http_buf);
}

void AppNode::Respond201(int fd, const char *content_type, string txt)
{
    stringstream rsp_stream;
    rsp_stream << 
        "HTTP/1.1 201 Created\r\n"
        "Server: TnmcAppNode/1.0\r\n"
        "Cache_Control: no-cache\r\n"
        "Content-Type: " << content_type << "\r\n"
        "Content-Length: " << txt.length() << "\r\n\r\n" << txt;

    networker.SendString(fd, rsp_stream.str());
}

void AppNode::Respond206(int fd, fw_file_size_t start, fw_file_size_t end, fw_file_size_t filelen)
{
    stringstream rsp_stream;
    rsp_stream << 
        "HTTP/1.1 206 Partial Content\r\n"
        "Server: TnmcAppNode/1.0\r\n"
        "Content-Range: bytes " << start << "-" << end << "/" << filelen << "\r\n"
        "Content-Type: application/octet-stream\r\n"
        "Transfer-Encoding: chunked\r\n"
        "Cache_Control: no-cache\r\n\r\n" << 
        hex << (end - start + 1) << "\r\n";

    networker.SendString(fd, rsp_stream.str());
}

void AppNode::HandleTimeout() 
{
    //purge all unconnected sessions
    if (!connecting_sessions.empty()) {
        set<Session *>::iterator it = connecting_sessions.begin();
        while (it != connecting_sessions.end()) {
            DelSession(*it);
            it++;
        }
        connecting_sessions.clear();
    }

    show_rate(); 
    set_timeout(); 
}

uint64_t AppNode::get_time_ms()
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    uint64_t diff; 
    if (tv.tv_usec < start_time.tv_usec) {
        tv.tv_sec--;
        diff = tv.tv_usec + 1000000 - start_time.tv_usec; 
    } else {
        diff = tv.tv_usec - start_time.tv_usec; 
    }

    diff /= 1000;
    assert(tv.tv_sec >= start_time.tv_sec);
    diff += 1000 * (tv.tv_sec - start_time.tv_sec);
    return diff;
}

MsgQueue::MsgQueue()
{
    pthread_mutex_init(&mutex, NULL);
}

MsgQueue::~MsgQueue()
{
    //destroy each msg in queue
    for (size_t i = 0; i < msg_queue.size(); i++) {
        FwAppMsg *msg = msg_queue.front();
        delete msg;
        msg_queue.pop();
    }

    pthread_mutex_destroy(&mutex);
}

bool MsgQueue::is_msgempty()
{	
    bool ret = false;
    pthread_mutex_lock(&mutex);
    if (msg_queue.size() == 0) {
        ret = true;
    }
    pthread_mutex_unlock(&mutex);
    return ret;
}

void MsgQueue::enqueue_msg(FwAppMsg *data)
{
    pthread_mutex_lock(&mutex);
    msg_queue.push(data);
    pthread_mutex_unlock(&mutex);
}

FwAppMsg *MsgQueue::dequeue_msg()
{
    FwAppMsg *msg = 0; 
    pthread_mutex_lock(&mutex);

    if (msg_queue.size() > 0) {
        msg = msg_queue.front();
        msg_queue.pop();
    }

    pthread_mutex_unlock(&mutex);
    return msg;
}

void AppNode::ExecuteWorkingThread() {
    while (true) {
        FwAppMsg *msg = static_cast<FwAppMsg *>(work_queue.dequeue_msg());
        if (!msg) {
            if (check_timeout()) {
                HandleTimeout();
            } else {
                usleep(100);
            }
            continue;
        }
       
        if (!VerifySockNode(msg->sock_node)) { 
            delete msg;
            continue;
        }

        Session *session = static_cast<Session *>(msg->sock_node->data); 
        session->Lock();

        try {
            switch (msg->type) {
                case FW_APP_MSG_TYPE_CONNECTED:
                    HandleConnedMsg(msg);
                    break;
                case FW_APP_MSG_TYPE_RECV:
                    HandleRecvMsg(msg);
                    break;
                case FW_APP_MSG_TYPE_SEND:
                    HandleSendMsg(msg);
                    break;
                case FW_APP_MSG_TYPE_NETWORK_ERR:
                    HandleNetworkErrMsg(msg);
                    break;
                case FW_APP_MSG_TYPE_CLOSE:
                    HandleCloseMsg(msg);
                    break;
                default:
                    assert(0);
                    break;
            }

            session->UnLock();
            //assume msg is already deleted if no exception thrown

        } catch (int event_code) {
            bool delete_session = true;
            switch (event_code) {
                case TNMC_FW_EVENT_CODE_HTTP400:
                    logger.log("chunk parse", LOG_LEVEL_ERROR);
                    break;
                case TNMC_FW_EVENT_CODE_EGAIN:
                    delete_session = false;
                    break;
                case TNMC_FW_EVENT_CODE_NETWORK_ERR:
                    logger.log("network send error", LOG_LEVEL_ERROR);
                    break;
                case TNMC_FW_EVENT_CODE_CLOSE: 
                    logger.log("network close", LOG_LEVEL_ERROR);
                    break;
                case TNMC_FW_EVENT_CODE_COMPLETED:
                    break;
                default:
                    logger.log("internal error", LOG_LEVEL_ERROR);
                    break;
            }
            
            if (delete_session) {
                DelSession(session);
            } else {
                session->UnLock();
            }

            delete msg;
        }
    }
}

void* process_do_app(AppNode *app)
{
    app->ExecuteWorkingThread();
    return 0;
}

void* process_do_networking(AppNode *app)
{
    pthread_detach(pthread_self());
    app->ExecuteNetworkingThread();
    return 0;
}





